package com.shujia.flink.sql

import org.apache.flink.table.api.{EnvironmentSettings, TableEnvironment}

object Demo4BlackHole {
  def main(args: Array[String]): Unit = {
    val settings: EnvironmentSettings = EnvironmentSettings
      .newInstance()
      .inStreamingMode()
      //.inBatchMode()
      .build()
    /**
     * flink sql环境
     *
     */
    val table: TableEnvironment = TableEnvironment.create(settings)


    table.executeSql(
      """
        |
        |CREATE TABLE datagen (
        | id STRING,
        | name STRING,
        | age INT
        |) WITH (
        | 'connector' = 'datagen',
        | 'rows-per-second' = '5', -- 每秒生成的数据行数据
        | 'fields.id.length' = '5', --字段长度限制
        | 'fields.name.length'='3',
        | 'fields.age.min' ='1', -- 最小值
        | 'fields.age.max'='100' -- 最大值
        |)
        |
        |""".stripMargin)

    table.executeSql(
      """
        |
        |CREATE TABLE blackhole_table
        |WITH ('connector' = 'blackhole')
        |LIKE datagen (EXCLUDING ALL)
        |
        |""".stripMargin)

    table.executeSql(
      """
        |
        |insert into blackhole_table
        |select * from datagen
        |
        |""".stripMargin)


  }

}
